from collections import defaultdict
from datetime import date, datetime, timedelta
import datetime as dt
from typing import Callable
from itertools import product
from functools import lru_cache
from time import time
import multiprocessing
import random
from typing import List, Dict

import numpy as np
import matplotlib.pyplot as plt
from pandas._libs.tslibs import Timestamp
import seaborn as sns
from pandas import DataFrame, to_datetime
from deap import creator, base, tools, algorithms

from common.object import CArbBarData, OrderData, TradeData, BarData, TickData, CTickBarData, CBarData, CTickData
from common.utility import round_to
from common.constant import Interval, Exchange, Direction, Offset, Status
from DataManager import database_manager
import json
import os
import redis
import zlib, pickle
from multiprocessing.sharedctypes import RawArray
import ctypes
import mmap
import math


from strategies.base import (
    BacktestingMode,
    EngineType,
    STOPORDER_PREFIX,
    StopOrder,
    StopOrderStatus,
    SPREADORDER_PREFIX,
    SpreadOrder,
    CommissionRate
)
from strategies.template import CtaTemplate

from enum import Enum

sns.set_style("whitegrid")
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)

# def round_to(price: float, priceStep: float):
#     return priceStep * int(price / priceStep + 0.000001)

class IterateMode(Enum):
    NONE = ""
    MONTHLY = "monthly"


class OptimizationSetting:
    """
    Setting for runnning optimization.
    """

    def __init__(self):
        """"""
        self.params = {}
        self.target_name = ""
        self.partial_settings = []

    def add_default_parameter(
        self, name: str, value
    ):
        """"""

        self.params[name] = [value]

    def add_parameter_set(self, name: str, val_list):
        self.params[name] = val_list

    def add_parameter(
        self, name: str, start: float, end: float = None, step: float = None
    ):
        """"""
        if not end and not step:
            self.params[name] = [start]
            return

        if start >= end:
            print("参数优化起始点必须小于终止点")
            return

        if step <= 0:
            print("参数优化步进必须大于0")
            return

        value = start
        value_list = []

        while value <= end:
            value_list.append(value)
            value += step

        self.params[name] = value_list

    def set_target(self, target_name: str):
        """"""
        self.target_name = target_name

    def generate_setting(self):
        """"""
        if len(self.params) > 0:
            keys = self.params.keys()
            values = self.params.values()
            products = list(product(*values))
        else:
            products = []

        settings = self.partial_settings
        for p in products:
            setting = dict(zip(keys, p))
            settings.append(setting)
        
        self.params.clear()

        return settings
    
    def generate_setting_ga(self):
        """""" 
        settings_ga = []
        settings = self.generate_setting()     
        for d in settings:            
            param = [tuple(i) for i in d.items()]
            settings_ga.append(param)
        return settings_ga


class BacktestingEngine:
    """"""

    engine_type = EngineType.BACKTESTING
    gateway_name = "BACKTESTING"

    def __init__(self):
        """"""
        self.vt_symbols = list()
        self.symbols = list()
        self.exchanges = list()
        self.start = None
        self.end = None
        self.rate = CommissionRate()
        self.slippage = 0
        self.size = 1
        self.pricetick = 0
        self.capital = 1_000_000
        self.mode = BacktestingMode.BAR

        self.strategy_class = None
        self.strategy = None
        self.ticks: Dict[str, CTickData] = dict()
        self.bars: Dict[str, CBarData] = dict()
        self.arbBars: Dict[str, CArbBarData] = dict()
        self.timestamp = None
        self.trading_day = None

        self.interval = None
        self.trading_start_date = 19700101
        self.tick_callback = None
        self.bar_callback = None
        self.history_data = None
        self.history_data_len = None
        self.history_bar_data = dict()

        self.stop_order_count = 0
        self.stop_orders = {}
        self.active_stop_orders = {}

        self.limit_order_count = 0
        self.limit_orders = {}
        self.active_limit_orders = {}

        self.spread_order_count = 0
        self.spread_orders = {}
        self.active_spread_orders = {}

        self.trade_count = 0
        self.trades = {}

        self.logs = []

        self.daily_results = {}
        self.daily_df = None
        self.sub_daily_df = {}
        self.redis_conn = redis.Redis(password="jingpyu-home")
        self.iteration_mode = IterateMode.NONE
        self.max_opens_per_time = -1
        self.active_arb_mode = False

    def clear_data(self):
        """
        Clear all data of last backtesting.
        """
        self.strategy = None
        self.ticks.clear()
        self.bars.clear()
        self.timestamp = None
        self.trading_day = None

        self.stop_order_count = 0
        self.stop_orders.clear()
        self.active_stop_orders.clear()

        self.spread_order_count = 0
        self.spread_orders.clear()

        self.limit_order_count = 0
        self.limit_orders.clear()
        self.active_limit_orders.clear()

        self.trade_count = 0
        self.trades.clear()

        self.logs.clear()
        self.daily_results.clear()

    def lock_cache(self, key, timeout=1800):
        lock_key = "lock_" + key
        return self.redis_conn.lock(lock_key, timeout=timeout)

    def store_cache(self, key: str, data, expire=20000):
        return self.redis_conn.set(key, zlib.compress(pickle.dumps(data), 5), ex=expire)

    def fetch_cache(self, key: str):
        redis_data = self.redis_conn.get(key)
        if redis_data is None:
            return None
        return pickle.loads(zlib.decompress(redis_data))

    def set_parameters(
        self,
        vt_symbols: List[str],
        exchanges: List[Exchange],
        interval: Interval,
        start: datetime,
        rate: CommissionRate,
        slippage: float,
        size: float,
        pricetick: float,
        capital: int = 0,
        end: datetime = None,
        mode: BacktestingMode = BacktestingMode.BAR,
        iterate_mode : IterateMode = IterateMode.NONE,
        active_arb_mode : bool = False
    ):
        """"""
        self.mode = mode
        self.vt_symbols = vt_symbols
        self.symbols = vt_symbols
        self.exchanges = exchanges
        self.interval = Interval(interval)
        self.rate = rate
        self.slippage = slippage
        self.size = size
        self.pricetick = pricetick
        self.start = start
        self.iteration_mode = iterate_mode
        self.active_arb_mode = active_arb_mode


        if capital:
            self.capital = capital

        if end:
            self.end = end

        if mode:
            self.mode = mode

    def add_strategy(self, strategy_class: type, setting: dict):
        """"""
        self.strategy_class = strategy_class
        self.strategy = strategy_class(
            self, strategy_class.__name__, self.vt_symbols, setting
        )

    def set_max_opens(self, num):
        self.max_opens_per_time = num

    def load_data(self, use_process=False, mmap_file="share.shm"):
        """"""
        self.output("开始加载历史数据 start: {} end: {}".format(self.start, self.end))

        if not self.end:
            self.end = datetime.now()

        if self.start >= self.end:
            self.output("起始日期必须小于结束日期")    
            return        

        self.history_bar_data.clear()
        if not use_process:
            self.history_bar_data, history_data = load_engine_data(self.start, self.end, tuple(self.symbols), tuple(self.exchanges), self.mode, self.interval, self.active_arb_mode)
            self.history_data = RawArray(CTickBarData, history_data)
            del history_data
        else:
            with multiprocessing.Manager() as manager:
                q = manager.list()
                q.append(mmap_file)

                fd = os.open(mmap_file, os.O_CREAT | os.O_TRUNC | os.O_RDWR)
                os.write(fd, b'\0' * ctypes.sizeof(CTickBarData))
                trans_map = mmap.mmap(fd, 0)
                
                p = multiprocessing.Process(target=load_engine_data_in_process, args=(q, self.start, self.end, tuple(self.symbols), tuple(self.exchanges), self.mode, self.interval, self.active_arb_mode))
                p.start()
                p.join()
                self.history_bar_data = q[1]
                data_len = q[2]
                self.history_data = RawArray(CTickBarData, data_len)
                trans_map.resize(ctypes.sizeof(CTickBarData) * data_len)
                ctypes.memmove(ctypes.addressof(self.history_data[0]), ctypes.addressof(CTickBarData.from_buffer(trans_map)), ctypes.sizeof(CTickBarData) * data_len)
                trans_map.close()
                del trans_map
                p.close()
                os.close(fd)
                os.remove(mmap_file)
        
        self.output(f"历史数据加载完成，数据量：{len(self.history_data)}")

    def load_history_data(self, vt_symbol, all=False):
        symbol, _ = vt_symbol.split(".")
        if symbol in self.history_bar_data:
            if all:
                return self.history_bar_data[symbol].copy()
            else:
                result = []
                for bar in self.history_bar_data[symbol]:
                    if bar.timestamp <= self.timestamp:
                        result.append(bar)
                return result
        else:
            return []

    def load_history_arb_data(self, all=False):
        import copy

        if self.mode != BacktestingMode.ArbMIXED:
            return []
        result = []
        for elem in self.history_data:
            if elem.type != b'arbBar':
                continue
            arbBar = elem.data.arbBar
            if all or arbBar.timestamp <= self.timestamp:
                result.append(copy.deepcopy(arbBar))
            
        return result


    def copy_data(self, other_engine):
        self.history_bar_data = other_engine.history_bar_data
        self.history_data = other_engine.history_data
        self.history_data_len = other_engine.history_data_len


    def run_backtesting(self):
        """"""
        self.strategy.on_init()

        # Use the first [days] of history data for initializing strategy
        ix = 0
        
        if self.history_data_len is None:
            self.history_data_len = len(self.history_data)

        for ix in range(self.history_data_len):
            data = self.history_data[ix]
            if data.type == b"bar":
                self.timestamp = data.data.bar.timestamp
                self.trading_day = data.data.bar.trading_day
            elif data.type == b'tick':
                self.timestamp = data.data.tick.timestamp
                self.trading_day = data.data.tick.trading_day
            else:
                self.timestamp = data.data.arbBar.timestamp
                self.trading_day = data.data.arbBar.trading_day
            if self.trading_day >= self.trading_start_date:
                break
            if data.type == b"bar":
                if self.bar_callback:
                    self.bar_callback(data.data.bar)
            elif data.type == b'tick':
                if self.tick_callback:
                    self.tick_callback(data.data.tick)

        self.strategy.set_inited(True)
        self.output("策略初始化完成")

        self.strategy.on_start()
        self.strategy.set_trading(True)
        self.output("开始回放历史数据")

        # Use the rest of history data for running backtesting
        for idx in range(ix, self.history_data_len):
            data = self.history_data[idx]
            if data.type == b"bar":
                self.new_bar(data.data.bar)
            elif data.type == b'tick':
                self.new_tick(data.data.tick)
            else:
                self.new_arbBar(data.data.arbBar)

        # 回测结束平掉所有头寸
        self.close_position()

        self.output("历史数据回放结束")

    def close_position(self, force=True, sub_strategy_id=None):
        for sub_id in self.strategy.pos_dict.keys():
            if sub_strategy_id is not None and sub_id != sub_strategy_id:
                continue
            if not force:
                # if there is any contract hit limit up/down, return false
                for vt_symbol, pos in self.strategy.pos_dict[sub_id].items():
                    if pos == 0:
                        continue
                    if self.mode != BacktestingMode.BAR and self.mode != BacktestingMode.ArbMIXED and (math.isclose(self.ticks[vt_symbol].ask_volume_1, 0) or math.isclose(self.ticks[vt_symbol].bid_volume_1, 0)):
                        return False
            
            for vt_symbol, pos in self.strategy.pos_dict[sub_id].items():
                if pos == 0:
                    continue

                self.limit_order_count += 1

                symbol, exchange = vt_symbol.split(".")

                if self.mode == BacktestingMode.BAR or self.mode == BacktestingMode.ArbMIXED:
                    close_price = self.bars[vt_symbol].close_price
                else:
                    symbol_tick = self.ticks[vt_symbol]
                    close_price = symbol_tick.last_price
                    if pos > 0 and symbol_tick.bid_volume_1 > 0:
                        close_price = symbol_tick.bid_price_1
                    elif pos < 0 and symbol_tick.ask_volume_1 > 0:
                        close_price = symbol_tick.ask_price_1

                datetime = dt.datetime.fromtimestamp(self.timestamp)
                order = OrderData(
                    symbol=symbol,
                    exchange=Exchange(exchange),
                    orderid=str(self.limit_order_count),
                    direction=Direction.LONG if pos > 0 else Direction.SHORT,
                    offset=Offset.CLOSE,
                    price=close_price,
                    volume=abs(pos),
                    status=Status.ALLTRADED,
                    gateway_name=self.gateway_name,
                )
                order.datetime = datetime
                order.trading_day = self.trading_day
                order.sub_strategy_id = sub_id

                self.limit_orders[order.vt_orderid] = order

                # Create trade data.
                self.trade_count += 1

                trade = TradeData(
                    symbol=symbol,
                    exchange=Exchange(exchange),
                    orderid=order.orderid,
                    tradeid=str(self.trade_count),
                    direction=order.direction,
                    offset=order.offset,
                    price=order.price,
                    volume=order.volume,
                    time=datetime.strftime("%H:%M:%S"),
                    gateway_name=self.gateway_name,
                    datetime = datetime,
                    trading_day = self.trading_day
                )
                trade.sub_strategy_id = order.sub_strategy_id

                self.trades[trade.vt_tradeid] = trade

            for vt_symbol in self.strategy.pos_dict[sub_id].keys():
                self.strategy.pos_dict[sub_id][vt_symbol] = 0
        
        return True
        

    class EngineEncoder(json.JSONEncoder):
        
        def default(self, o):
            if isinstance(o, TradeData):
                return {"symbol": o.symbol, "tradeid": o.tradeid, "direction": o.direction.name, "offset": o.offset.name, "price": o.price, "volume": o.volume, "datetime": o.datetime.strftime("%Y%m%d %H:%M:%S.%f"), "exchange": o.exchange.name, "gateway_name": o.gateway_name, "orderid": o.orderid, "trading_day": o.trading_day, "sub_sid": o.sub_strategy_id}
            elif isinstance(o, datetime):
                return o.strftime("%Y%m%d %H:%M:%S.%f")
            elif isinstance(o, date):
                return o.strftime("%Y%m%d")
            elif isinstance(o, np.int64):
                return int(o)
            return super().default(o)

    def calculate_result(self):
        """"""
        self.output("开始计算逐日盯市盈亏")

        import copy
        # if not self.trades:
        #     self.output("成交记录为空，无法计算")
        #     return None

        orig_daily_results = copy.deepcopy(self.daily_results)

        # generate default df for every sub strategy
        results = defaultdict(list)

        for daily_result in self.daily_results.values():
            for key, value in daily_result.__dict__.items():
                results[key].append(value)

        empty_df = DataFrame.from_dict(results).set_index("date")
        empty_df.index = to_datetime(empty_df.index, format="%Y%m%d")

        self.sub_daily_df[None] = empty_df
        for i in range(100):
            self.sub_daily_df[i] = empty_df

        # Add trade data into daily reuslt.
        sub_ids = set()
        for trade in self.trades.values():
            d = trade.trading_day
            daily_result = self.daily_results[d]
            daily_result.add_trade(trade)
            sub_ids.add(trade.sub_strategy_id)

        # Calculate daily result by iteration.
        pre_close = defaultdict(float)
        start_pos = defaultdict(int)

        for daily_result in self.daily_results.values():
            for vt_symbol in daily_result.close_price_dict.keys():
                daily_result.calculate_pnl(
                    vt_symbol, pre_close[vt_symbol], start_pos[vt_symbol], self.size, self.rate, self.slippage
                )

                pre_close[vt_symbol] = daily_result.close_price_dict[vt_symbol]
                start_pos[vt_symbol] = daily_result.end_pos_dict[vt_symbol]

        # Generate dataframe
        results = defaultdict(list)

        for daily_result in self.daily_results.values():
            for key, value in daily_result.__dict__.items():
                results[key].append(value)

        self.daily_df = DataFrame.from_dict(results).set_index("date")
        self.daily_df.index = to_datetime(self.daily_df.index, format="%Y%m%d")

        for sub_id in sub_ids:
            daily_results = copy.deepcopy(orig_daily_results)
            for trade in self.trades.values():
                if trade.sub_strategy_id != sub_id:
                    continue
                d = trade.trading_day
                daily_result = daily_results[d]
                daily_result.add_trade(trade)

            # Calculate daily result by iteration.
            pre_close = defaultdict(float)
            start_pos = defaultdict(int)

            for daily_result in daily_results.values():
                for vt_symbol in daily_result.close_price_dict.keys():
                    daily_result.calculate_pnl(
                        vt_symbol, pre_close[vt_symbol], start_pos[vt_symbol], self.size, self.rate, self.slippage
                    )

                    pre_close[vt_symbol] = daily_result.close_price_dict[vt_symbol]
                    start_pos[vt_symbol] = daily_result.end_pos_dict[vt_symbol]

            # Generate dataframe
            results = defaultdict(list)

            for daily_result in daily_results.values():
                for key, value in daily_result.__dict__.items():
                    results[key].append(value)

            df = DataFrame.from_dict(results).set_index("date")
            df.index = to_datetime(df.index, format="%Y%m%d")
            self.sub_daily_df[sub_id] = df

        self.output("逐日盯市盈亏计算完成")
        return self.daily_df

    def calculate_statistics(self, df: DataFrame = None, output=True):
        """"""
        if output:
            self.output("开始计算策略统计指标")

        # Check DataFrame input exterior
        if df is None:
            df = self.daily_df
            if self.strategy.get_strategy_group_num() > 1:
                for sub_strategy in self.strategy.strategy_list:
                    sub_sid = sub_strategy.sub_strategy_id
                    sub_df = self.sub_daily_df[sub_sid]
                    self.calculate_statistics(sub_df, False)
        
        # Check for init DataFrame 
        if df is None:
            # Set all statistics to 0 if no trade.
            start_date = ""
            end_date = ""
            total_days = 0
            profit_days = 0
            loss_days = 0
            end_balance = 0
            max_drawdown = 0
            max_ddpercent = 0
            total_net_pnl = 0
            daily_net_pnl = 0
            total_commission = 0
            daily_commission = 0
            total_slippage = 0
            daily_slippage = 0
            total_turnover = 0
            daily_turnover = 0
            total_trade_count = 0
            daily_trade_count = 0
            total_return = 0
            annual_return = 0
            daily_return = 0
            return_std = 0
            sharpe_ratio = 0
            return_drawdown_ratio = 0
            profit_per_trade = 0
        else:
            # Calculate balance related time series data
            df.loc[:, "balance"] = df["net_pnl"].cumsum() + self.capital
            df.loc[:, "return"] = np.log(df["balance"] / df["balance"].shift(1)).fillna(0)
            df.loc[:, "highlevel"] = (
                df["balance"].rolling(
                    min_periods=1, window=len(df), center=False).max()
            )
            df.loc[:, "drawdown"] = df["balance"] - df["highlevel"]
            df.loc[:, "ddpercent"] = df["drawdown"] / df["highlevel"] * 100

            # Calculate statistics value
            start_date = df.index[0]
            end_date = df.index[-1]

            total_days = len(df)
            profit_days = len(df[df["net_pnl"] > 0])
            loss_days = len(df[df["net_pnl"] < 0])

            end_balance = df["balance"].iloc[-1]
            max_drawdown = df["drawdown"].min()
            max_ddpercent = df["ddpercent"].min()

            total_net_pnl = df["net_pnl"].sum()
            daily_net_pnl = total_net_pnl / total_days

            total_commission = df["commission"].sum()
            daily_commission = total_commission / total_days

            total_slippage = df["slippage"].sum()
            daily_slippage = total_slippage / total_days

            total_turnover = df["turnover"].sum()
            daily_turnover = total_turnover / total_days

            total_trade_count = df["trade_count"].sum()
            daily_trade_count = total_trade_count / total_days

            total_return = (end_balance / self.capital - 1) * 100
            annual_return = total_return / total_days * 240
            daily_return = df["return"].mean() * 100
            return_std = df["return"].std() * 100
            if total_trade_count > 0:
                profit_per_trade = total_net_pnl / total_trade_count
            else:
                profit_per_trade = 0

            if return_std:
                sharpe_ratio = daily_return / return_std * np.sqrt(240)
            else:
                sharpe_ratio = 0

            if math.isclose(max_ddpercent, 0):
                return_drawdown_ratio = 100
            else:
                return_drawdown_ratio = -total_return / max_ddpercent

        # Output
        if output:
            self.output("-" * 30)
            self.output(f"首个交易日：\t{start_date}")
            self.output(f"最后交易日：\t{end_date}")

            self.output(f"总交易日：\t{total_days}")
            self.output(f"盈利交易日：\t{profit_days}")
            self.output(f"亏损交易日：\t{loss_days}")

            self.output(f"起始资金：\t{self.capital:,.2f}")
            self.output(f"结束资金：\t{end_balance:,.2f}")

            self.output(f"总收益率：\t{total_return:,.2f}%")
            self.output(f"年化收益：\t{annual_return:,.2f}%")
            self.output(f"最大回撤: \t{max_drawdown:,.2f}")
            self.output(f"百分比最大回撤: {max_ddpercent:,.2f}%")

            self.output(f"总盈亏：\t{total_net_pnl:,.2f}")
            self.output(f"总手续费：\t{total_commission:,.2f}")
            self.output(f"总滑点：\t{total_slippage:,.2f}")
            self.output(f"总成交金额：\t{total_turnover:,.2f}")
            self.output(f"总成交笔数：\t{total_trade_count}")

            self.output(f"日均盈亏：\t{daily_net_pnl:,.2f}")
            self.output(f"日均手续费：\t{daily_commission:,.2f}")
            self.output(f"日均滑点：\t{daily_slippage:,.2f}")
            self.output(f"日均成交金额：\t{daily_turnover:,.2f}")
            self.output(f"日均成交笔数：\t{daily_trade_count}")

            self.output(f"日均收益率：\t{daily_return:,.2f}%")
            self.output(f"收益标准差：\t{return_std:,.2f}%")
            self.output(f"Sharpe Ratio：\t{sharpe_ratio:,.2f}")
            self.output(f"收益回撤比：\t{return_drawdown_ratio:,.2f}")
            self.output(f"每笔盈亏：\t{profit_per_trade:,.2f}")

        statistics = {
            "start_date": start_date,
            "end_date": end_date,
            "total_days": total_days,
            "profit_days": profit_days,
            "loss_days": loss_days,
            "capital": self.capital,
            "end_balance": end_balance,
            "max_drawdown": max_drawdown,
            "max_ddpercent": max_ddpercent,
            "total_net_pnl": total_net_pnl,
            "daily_net_pnl": daily_net_pnl,
            "total_commission": total_commission,
            "daily_commission": daily_commission,
            "total_slippage": total_slippage,
            "daily_slippage": daily_slippage,
            "total_turnover": total_turnover,
            "daily_turnover": daily_turnover,
            "total_trade_count": total_trade_count,
            "daily_trade_count": daily_trade_count,
            "total_return": total_return,
            "annual_return": annual_return,
            "daily_return": daily_return,
            "return_std": return_std,
            "sharpe_ratio": sharpe_ratio,
            "return_drawdown_ratio": return_drawdown_ratio,
            "profit_per_trade": profit_per_trade,
        }

        return statistics

    @staticmethod
    def show_chart(df, save_file:str = None):
        """"""
        # Check for init DataFrame        
        if df is None:
            return

        plt.figure(figsize=(10, 16))

        balance_plot = plt.subplot(4, 1, 1)
        balance_plot.set_title("Balance")
        blance_axes = df["balance"].plot(legend=True)
        balance_plot.get_yaxis().get_major_formatter().set_scientific(False)
        balance_plot.get_yaxis().get_major_formatter().set_useOffset(False)

        drawdown_plot = plt.subplot(4, 1, 2)
        drawdown_plot.set_title("Drawdown")
        drawdown_plot.fill_between(range(len(df)), df["drawdown"].values)

        pnl_plot = plt.subplot(4, 1, 3)
        pnl_plot.set_title("Daily Pnl")
        df["net_pnl"].plot(kind="bar", legend=False, grid=False, xticks=[])

        distribution_plot = plt.subplot(4, 1, 4)
        distribution_plot.set_title("Daily Pnl Distribution")
        df["net_pnl"].hist(bins=50)
        if save_file:
            plt.savefig(save_file)
        else:
            plt.show()
        plt.close('all')

    def save_backtest_result(self, dir: str, daily_df = None, trades=None, logs=None):
        if not os.path.exists(dir) or not os.path.isdir(dir):
            os.makedirs(dir)
        fig_file = os.path.join(dir, "result.png")
        daily_file = os.path.join(dir, "daily_df.csv")
        if daily_df is None:
            self.show_chart(self.daily_df, fig_file)
            self.daily_df.to_csv(daily_file)
        else:
            self.show_chart(daily_df, fig_file)
            daily_df.to_csv(daily_file)

        if self.mode == BacktestingMode.ArbMIXED and self.history_data is not None:
            with open(os.path.join(dir, "arb_bars.csv"), "w") as f:
                for data in self.history_data:
                    if data.type == b"arbBar":
                        arbBar = data.data.arbBar
                        dt_obj = datetime.fromtimestamp(arbBar.timestamp)
                        line = dt_obj.strftime("%Y%m%d %X") + "," + arbBar.symbol1.decode('utf-8') + "," + arbBar.symbol2.decode('utf-8') + ","
                        line = line + str(arbBar.bid_ticks) + "," + str(arbBar.bid_open_price) + "," + str(arbBar.bid_high_price) + "," + str(arbBar.bid_low_price) + "," + str(arbBar.bid_low_price) + ","
                        line = line + str(arbBar.ask_ticks) + "," + str(arbBar.ask_open_price) + "," + str(arbBar.ask_high_price) + "," + str(arbBar.ask_low_price) + "," + str(arbBar.ask_low_price)
                        f.write(line + "\r\n")

        if trades is None and daily_df is None:
            with open(os.path.join(dir, "trades.json"), "w") as f:
                json.dump(self.trades, f, cls=self.EngineEncoder, indent=2)
        elif trades is not None:
            with open(os.path.join(dir, "trades.json"), "w") as f:
                json.dump(trades, f, cls=self.EngineEncoder, indent=2)

        if logs is None and daily_df is None:
            with open(os.path.join(dir, "logs.txt"), "w") as f:
                for line in self.logs:
                    if line is not None:
                        f.write(line + "\r\n")
        elif logs is not None:
            with open(os.path.join(dir, "logs.txt"), "w") as f:
                for line in logs:
                    if line is not None:
                        f.write(line + "\r\n")

        if self.strategy.get_strategy_group_num() > 1 and daily_df is None and trades is None and logs is None:
            for sub_strategy in self.strategy.strategy_list:
                sub_sid = sub_strategy.sub_strategy_id
                fig_file = os.path.join(dir, f"result_{sub_sid}.png")
                daily_file = os.path.join(dir, f"daily_df_{sub_sid}.csv")
                self.show_chart(self.sub_daily_df[sub_sid], fig_file)
                self.sub_daily_df[sub_sid].to_csv(daily_file)

                sub_trades = {k:v for k,v in self.trades.items() if v.sub_strategy_id == sub_sid}
                with open(os.path.join(dir, f"trades_{sub_sid}.json"), "w") as f:
                    json.dump(sub_trades, f, cls=self.EngineEncoder, indent=2)

    def run_optimization(self, optimization_setting: OptimizationSetting, output=True, max_cores=999, start_date: date = None, parallel: int = 1):
        """"""
        # Get optimization setting and target
        settings = optimization_setting.generate_setting()
        target_name = optimization_setting.target_name

        if not settings:
            self.output("优化参数组合为空，请检查")
            return

        if not target_name:
            self.output("优化目标未设置，请检查")
            return

        import gc
        
        multiprocessing.set_start_method("spawn", True)
        if self.history_data is None or self.history_bar_data is None:
            self.load_data(True)

        # Use multiprocessing pool for running backtesting with different setting
        pool = multiprocessing.Pool(min(max_cores, multiprocessing.cpu_count()), initializer=init_process, initargs=(self.history_data,))

        results = []
        for setting in settings:
            result = (pool.apply_async(optimize, (
                target_name,
                self.strategy_class,
                setting,
                self.vt_symbols,
                self.interval,
                self.start,
                self.rate,
                self.slippage,
                self.size,
                self.pricetick,
                self.capital,
                self.end,
                self.mode,
                start_date,
                self.history_bar_data,
                self.iteration_mode,
                parallel,
                self.active_arb_mode
            )))
            results.append(result)

        pool.close()
        pool.join()

        # Sort results and output
        result_values = [result.get() for result in results]
        result_values.sort(reverse=True, key=lambda result: result[1])

        pool.terminate()

        if output:
            for value in result_values:
                msg = f"参数：{value[0]}, 目标：{value[1]}"
                self.output(msg)

        del self.history_bar_data 
        del self.history_data

        gc.collect()

        return result_values

    def run_ga_optimization(self, optimization_setting: OptimizationSetting, population_size=100, ngen_size=30, output=True):
        """"""
        # Get optimization setting and target
        settings = optimization_setting.generate_setting_ga()
        target_name = optimization_setting.target_name

        if not settings:
            self.output("优化参数组合为空，请检查")
            return

        if not target_name:
            self.output("优化目标未设置，请检查")
            return

        # Define parameter generation function
        def generate_parameter():
            """"""
            return random.choice(settings)
        
        def mutate_individual(individual, indpb):
            """"""
            size = len(individual)
            paramlist = generate_parameter()
            for i in range(size):
                if random.random() < indpb:
                    individual[i] = paramlist[i]
            return individual,

        # Create ga object function
        global ga_target_name
        global ga_strategy_class
        global ga_setting
        global ga_vt_symbols
        global ga_interval
        global ga_start
        global ga_rate
        global ga_slippage
        global ga_size
        global ga_pricetick
        global ga_capital
        global ga_end
        global ga_mode

        ga_target_name = target_name
        ga_strategy_class = self.strategy_class
        ga_setting = settings[0]
        ga_vt_symbols = self.vt_symbols
        ga_interval = self.interval
        ga_start = self.start
        ga_rate = self.rate
        ga_slippage = self.slippage
        ga_size = self.size
        ga_pricetick = self.pricetick
        ga_capital = self.capital
        ga_end = self.end
        ga_mode = self.mode

        # Set up genetic algorithem
        toolbox = base.Toolbox() 
        toolbox.register("individual", tools.initIterate, creator.Individual, generate_parameter)                          
        toolbox.register("population", tools.initRepeat, list, toolbox.individual)                                            
        toolbox.register("mate", tools.cxTwoPoint)                                               
        toolbox.register("mutate", mutate_individual, indpb=1)               
        toolbox.register("evaluate", ga_optimize)                                                
        toolbox.register("select", tools.selNSGA2)       

        total_size = len(settings)
        pop_size = population_size                      # number of individuals in each generation
        lambda_ = pop_size                              # number of children to produce at each generation
        mu = int(pop_size * 0.8)                        # number of individuals to select for the next generation

        cxpb = 0.95         # probability that an offspring is produced by crossover    
        mutpb = 1 - cxpb    # probability that an offspring is produced by mutation
        ngen = ngen_size    # number of generation
                
        pop = toolbox.population(pop_size)      
        hof = tools.ParetoFront()               # end result of pareto front

        stats = tools.Statistics(lambda ind: ind.fitness.values)
        np.set_printoptions(suppress=True)
        stats.register("mean", np.mean, axis=0)
        stats.register("std", np.std, axis=0)
        stats.register("min", np.min, axis=0)
        stats.register("max", np.max, axis=0)

        # Multiprocessing is not supported yet.
        # pool = multiprocessing.Pool(multiprocessing.cpu_count())
        # toolbox.register("map", pool.map)

        # Run ga optimization
        self.output(f"参数优化空间：{total_size}")
        self.output(f"每代族群总数：{pop_size}")
        self.output(f"优良筛选个数：{mu}")
        self.output(f"迭代次数：{ngen}")
        self.output(f"交叉概率：{cxpb:.0%}")
        self.output(f"突变概率：{mutpb:.0%}")

        start = time()

        algorithms.eaMuPlusLambda(
            pop, 
            toolbox, 
            mu, 
            lambda_, 
            cxpb, 
            mutpb, 
            ngen, 
            stats,
            halloffame=hof
        )    
        
        end = time()
        cost = int((end - start))

        self.output(f"遗传算法优化完成，耗时{cost}秒")
        
        # Return result list
        results = []

        for parameter_values in hof:
            setting = dict(parameter_values)
            target_value = ga_optimize(parameter_values)[0]
            results.append((setting, target_value, {}))
        
        return results

    def update_daily_close(self, price: float, vt_symbol: str, trading_day: date):
        """"""
        daily_result = self.daily_results.get(trading_day, None)
        if daily_result:
            daily_result.close_price_dict[vt_symbol] = price
        else:
            self.daily_results[trading_day] = DailyResult(trading_day)
            self.daily_results[trading_day].close_price_dict[vt_symbol] = price

    def new_arbBar(self, arbBar: CArbBarData):
        """"""
        vt_symbol = arbBar.vt_symbol1.decode('utf-8') + "&" + arbBar.vt_symbol2.decode('utf-8')
        self.timestamp = arbBar.timestamp
        self.arbBars[vt_symbol] = arbBar
            
        if self.trading_day != arbBar.trading_day:
            self.strategy.on_new_day(arbBar.trading_day)
        self.trading_day = arbBar.trading_day
        
        self.cross_spread_order(vt_symbol)
        self.strategy.on_arb_bar(arbBar)

    def new_bar(self, bar: CBarData):
        """"""
        vt_symbol = bar.vt_symbol.decode('utf-8')
        bar_tradingDay = bar.trading_day
        if self.mode == BacktestingMode.BAR or self.mode == BacktestingMode.ArbMIXED:
            if self.iteration_mode == IterateMode.MONTHLY and ((self.trading_day % 10000) // 100) != ((bar_tradingDay % 10000) // 100):
                self.close_position()
                self.strategy.on_iteration_end(self.trading_day)

        self.bars[vt_symbol] = bar
        if self.mode == BacktestingMode.BAR or self.mode == BacktestingMode.ArbMIXED:
            self.timestamp = bar.timestamp
            
            if self.trading_day != bar_tradingDay:
                self.strategy.on_new_day(bar_tradingDay)
            self.trading_day = bar_tradingDay
            
            self.cross_limit_order(vt_symbol)
            self.cross_stop_order(vt_symbol)
        self.strategy.on_bar(bar)
        
        if self.mode == BacktestingMode.BAR or self.mode == BacktestingMode.ArbMIXED:
            self.update_daily_close(bar.close_price, vt_symbol, bar_tradingDay)

    def new_tick(self, tick: CTickData):
        """"""
        vt_symbol = tick.vt_symbol.decode('utf-8')
        tick_tradingDay = tick.trading_day
        if self.iteration_mode == IterateMode.MONTHLY and ((self.trading_day % 10000) // 100) != ((tick_tradingDay % 10000) // 100):
            self.close_position()
            self.strategy.on_iteration_end(self.trading_day)

        self.ticks[vt_symbol] = tick
        self.timestamp = tick.timestamp

        if self.trading_day != tick_tradingDay:
            self.strategy.on_new_day(tick_tradingDay)
        self.trading_day = tick_tradingDay

        self.cross_limit_order(vt_symbol)
        self.cross_stop_order(vt_symbol)
        self.cross_spread_order(vt_symbol)
        self.strategy.on_tick(tick)

        self.update_daily_close((tick.bid_price_1 + tick.ask_price_1)/2 if tick.ask_volume_1 > 0 and tick.bid_volume_1 > 0 else tick.last_price, vt_symbol, tick_tradingDay)

    def cross_limit_order(self, vt_symbol: str):
        """
        Cross limit order with last bar/tick data.
        """
        if len(self.active_limit_orders) == 0:
            return

        if self.mode == BacktestingMode.BAR:
            bar = self.bars[vt_symbol]
            long_cross_price = bar.low_price + self.pricetick
            short_cross_price = bar.high_price - self.pricetick
            long_best_price = bar.open_price
            short_best_price = bar.open_price
        else:
            tick = self.ticks[vt_symbol]
            long_cross_price = tick.ask_price_1 if tick.ask_volume_1 > 0 else 0
            short_cross_price = tick.bid_price_1 if tick.bid_volume_1 > 0 else 0
            long_best_price = long_cross_price
            short_best_price = short_cross_price

        for order in list(self.active_limit_orders.values()):
            if order.vt_symbol != vt_symbol:
                continue

            first_cross = False
            # Push order update with status "not traded" (pending).
            if order.status == Status.SUBMITTING:
                order.status = Status.NOTTRADED
                self.strategy.on_order(order)
                first_cross = True

            # Check whether limit orders can be filled.
            long_cross = (
                ((order.direction == Direction.LONG and order.offset == Offset.OPEN) or (order.direction == Direction.SHORT and order.offset != Offset.OPEN))
                and (order.price >= long_cross_price or math.isclose(order.price, long_cross_price))
                and long_cross_price > 0
            )

            short_cross = (
                ((order.direction == Direction.LONG and order.offset != Offset.OPEN) or (order.direction == Direction.SHORT and order.offset == Offset.OPEN))
                and (order.price <= short_cross_price or math.isclose(order.price, short_cross_price))
                and short_cross_price > 0
            )

            if not long_cross and not short_cross:
                continue

            # Push order udpate with status "all traded" (filled).
            order.traded = order.volume
            order.status = Status.ALLTRADED
            self.strategy.on_order(order)

            self.active_limit_orders.pop(order.vt_orderid)

            datetime = dt.datetime.fromtimestamp(self.timestamp)
            # Push trade update
            self.trade_count += 1

            if long_cross:
                if first_cross:
                    trade_price = min(order.price, long_best_price)
                else:
                    trade_price = max(order.price, long_cross_price)
                pos_change = order.volume
            else:
                if first_cross:
                    trade_price = max(order.price, short_best_price)
                else:
                    trade_price = min(order.price, short_cross_price)
                pos_change = -order.volume

            trade = TradeData(
                symbol=order.symbol,
                exchange=order.exchange,
                orderid=order.orderid,
                tradeid=str(self.trade_count),
                direction=order.direction,
                offset=order.offset,
                price=trade_price,
                volume=order.volume,
                time=datetime.strftime("%H:%M:%S"),
                gateway_name=self.gateway_name,
                datetime = datetime,
                trading_day = self.trading_day
            )
            trade.sub_strategy_id = order.sub_strategy_id

            self.strategy.pos_dict[order.sub_strategy_id][vt_symbol] += pos_change
            self.strategy.on_trade(trade)

            self.trades[trade.vt_tradeid] = trade

    def cross_stop_order(self, vt_symbol: str):
        """
        Cross stop order with last bar/tick data.
        """
        if len(self.active_stop_orders) == 0:
            return

        if self.mode == BacktestingMode.BAR:
            bar = self.bars[vt_symbol]
            long_cross_price = bar.high_price
            short_cross_price = bar.low_price
            long_best_price = bar.open_price
            short_best_price = bar.open_price
            symbol = bar.symbol
            exchange = bar.exchange
        else:
            tick = self.ticks[vt_symbol]
            long_cross_price = tick.last_price
            short_cross_price = tick.last_price
            long_best_price = long_cross_price
            short_best_price = short_cross_price
            symbol = tick.symbol
            exchange = tick.exchange

        for stop_order in list(self.active_stop_orders.values()):
            if stop_order.vt_symbol != vt_symbol:
                continue
            # Check whether stop order can be triggered.
            long_cross = (
                stop_order.direction == Direction.LONG 
                and stop_order.price <= long_cross_price
            )

            short_cross = (
                stop_order.direction == Direction.SHORT 
                and stop_order.price >= short_cross_price
            )

            if not long_cross and not short_cross:
                continue
            
            datetime = dt.datetime.fromtimestamp(self.timestamp)
            # Create order data.
            self.limit_order_count += 1

            order = OrderData(
                symbol=symbol,
                exchange=exchange,
                orderid=str(self.limit_order_count),
                direction=stop_order.direction,
                offset=stop_order.offset,
                price=stop_order.price,
                volume=stop_order.volume,
                status=Status.ALLTRADED,
                gateway_name=self.gateway_name,
            )
            order.datetime = datetime
            order.trading_day = self.trading_day
            order.sub_strategy_id = stop_order.sub_strategy_id

            self.limit_orders[order.vt_orderid] = order

            # Create trade data.
            if long_cross:
                trade_price = max(stop_order.price, long_best_price)
                pos_change = order.volume
            else:
                trade_price = min(stop_order.price, short_best_price)
                pos_change = -order.volume

            self.trade_count += 1

            trade = TradeData(
                symbol=order.symbol,
                exchange=order.exchange,
                orderid=order.orderid,
                tradeid=str(self.trade_count),
                direction=order.direction,
                offset=order.offset,
                price=trade_price,
                volume=order.volume,
                time=datetime.strftime("%H:%M:%S"),
                gateway_name=self.gateway_name,
                datetime = datetime,
                trading_day = self.trading_day
            )
            trade.sub_strategy_id = order.sub_strategy_id

            self.trades[trade.vt_tradeid] = trade

            # Update stop order.
            stop_order.vt_orderid = order.vt_orderid
            stop_order.status = StopOrderStatus.TRIGGERED

            self.active_stop_orders.pop(stop_order.stop_orderid)

            # Push update to strategy.
            self.strategy.on_stop_order(stop_order)
            self.strategy.on_order(order)

            self.strategy.pos_dict[stop_order.sub_strategy_id][vt_symbol] += pos_change
            self.strategy.on_trade(trade)

    def cross_spread_order(self, vt_symbol):
        """
        Cross spread order with last bar/tick data.
        """
        if self.mode == BacktestingMode.BAR:
            return

        if len(self.active_spread_orders) == 0:
            return

        arb_mode = False
        arb_symbol1 = None
        arb_symbol2 = None
        if "&" in vt_symbol:
            # arb bar mode
            arb_mode = True

        valid_spread_orders = []
        for spread_order in list(self.active_spread_orders.values()):
            if not arb_mode:
                if spread_order.vt_symbol1 != vt_symbol and spread_order.vt_symbol2 != vt_symbol:
                    continue

                if spread_order.vt_symbol1 not in self.ticks or spread_order.vt_symbol2 not in self.ticks:
                    continue
            else:
                [arb_symbol1, arb_symbol2] = vt_symbol.split("&")
                if spread_order.vt_symbol1 != arb_symbol1 or spread_order.vt_symbol2 != arb_symbol2:
                    continue

            valid_spread_orders.append(spread_order)
        
        if self.max_opens_per_time < 0:
            pending_spread_orders = valid_spread_orders
        else:
            pending_spread_orders = [order for order in valid_spread_orders if order.offset == Offset.CLOSE]
            open_long_orders = [order for order in valid_spread_orders if order.offset == Offset.OPEN and order.direction == Direction.LONG]
            sorted_list = sorted(open_long_orders, key=lambda s: s.price - 1e-6 * s.sub_strategy_id, reverse=True)
            pending_spread_orders.extend(sorted_list[:self.max_opens_per_time])

            open_short_orders = [order for order in valid_spread_orders if order.offset == Offset.OPEN and order.direction == Direction.SHORT]
            sorted_list = sorted(open_short_orders, key=lambda s: s.price + 1e-6 * s.sub_strategy_id, reverse=False)
            pending_spread_orders.extend(sorted_list[:self.max_opens_per_time])


        for spread_order in pending_spread_orders:
            # Check whether spread order can be triggered.
            long_cross = short_cross = False
            price1 = price2 = 0
            if (spread_order.direction == Direction.LONG and spread_order.offset == Offset.OPEN) or (spread_order.direction == Direction.SHORT and spread_order.offset != Offset.OPEN):
                if not arb_mode:
                    price1 = self.ticks[spread_order.vt_symbol1].ask_price_1
                    price2 = self.ticks[spread_order.vt_symbol2].bid_price_1
                    long_cross_price = price1 - price2
                    long_cross_condition = (self.ticks[spread_order.vt_symbol1].ask_volume_1 > 0) and (self.ticks[spread_order.vt_symbol2].bid_volume_1 > 0)
                    long_cross = (spread_order.price >= long_cross_price and long_cross_condition)
                else:
                    price1 = self.bars[spread_order.vt_symbol1].close_price
                    long_cross = (spread_order.price >= self.arbBars[vt_symbol].ask_low_price and self.arbBars[vt_symbol].ask_ticks > 0)
            else:
                if not arb_mode:
                    price1 = self.ticks[spread_order.vt_symbol1].bid_price_1
                    price2 = self.ticks[spread_order.vt_symbol2].ask_price_1
                    short_cross_price = price1 - price2
                    short_cross_condition = (self.ticks[spread_order.vt_symbol1].bid_volume_1 > 0) and (self.ticks[spread_order.vt_symbol2].ask_volume_1 > 0)
                    short_cross = (spread_order.price <= short_cross_price and short_cross_condition)
                else:
                    price1 = self.bars[spread_order.vt_symbol1].close_price
                    short_cross = (spread_order.price <= self.arbBars[vt_symbol].bid_high_price and self.arbBars[vt_symbol].bid_ticks > 0)

            if not long_cross and not short_cross:
                spread_order.continuous_cross = 0
                continue

            datetime = dt.datetime.fromtimestamp(self.timestamp)

            symbol, exchange = spread_order.vt_symbol1.split(".")
            if not arb_mode:
                if spread_order.continuous_cross == 0:
                    spread_order.continuous_cross = 1
                    continue

                spread_order.continuous_cross += 1

                if exchange == "SHFE" and spread_order.continuous_cross < 4:
                    continue

            # Create order data.
            self.limit_order_count += 1

            order = OrderData(
                symbol=symbol,
                exchange=Exchange(exchange),
                orderid=str(self.limit_order_count),
                direction=spread_order.direction,
                offset=spread_order.offset,
                price=price1,
                volume=spread_order.volume,
                status=Status.ALLTRADED,
                gateway_name=self.gateway_name,
            )
            order.datetime = datetime
            order.trading_day = self.trading_day
            order.sub_strategy_id = spread_order.sub_strategy_id

            self.limit_orders[order.vt_orderid] = order

            # Create trade data.
            self.trade_count += 1

            if long_cross:
                pos_change = order.volume
            else:
                pos_change = -order.volume

            trade = TradeData(
                symbol=symbol,
                exchange=Exchange(exchange),
                orderid=order.orderid,
                tradeid=str(self.trade_count),
                direction=order.direction,
                offset=order.offset,
                price=price1,
                volume=order.volume,
                time=datetime.strftime("%H:%M:%S"),
                gateway_name=self.gateway_name,
                datetime = datetime,
                trading_day = self.trading_day
            )
            trade.sub_strategy_id = order.sub_strategy_id

            self.trades[trade.vt_tradeid] = trade

            # Update spread order.
            spread_order.vt_orderids.append(order.vt_orderid)

            # Push update to strategy.
            self.strategy.on_order(order)

            self.strategy.pos_dict[spread_order.sub_strategy_id][spread_order.vt_symbol1] += pos_change
            self.strategy.on_trade(trade)

            # leg 2

            self.limit_order_count += 1

            trade_price = price1 - spread_order.price
            symbol, exchange = spread_order.vt_symbol2.split(".")

            order = OrderData(
                symbol=symbol,
                exchange=Exchange(exchange),
                orderid=str(self.limit_order_count),
                direction= (Direction.SHORT if Direction.LONG == spread_order.direction else Direction.LONG),
                offset=spread_order.offset,
                price=trade_price,
                volume=spread_order.volume,
                status=Status.ALLTRADED,
                gateway_name=self.gateway_name,
            )
            order.datetime = datetime
            order.trading_day = self.trading_day
            order.sub_strategy_id = spread_order.sub_strategy_id

            self.limit_orders[order.vt_orderid] = order

            # Create trade data.
            self.trade_count += 1

            if long_cross:
                pos_change = -order.volume
            else:
                pos_change = order.volume

            trade = TradeData(
                symbol=symbol,
                exchange=Exchange(exchange),
                orderid=order.orderid,
                tradeid=str(self.trade_count),
                direction=order.direction,
                offset=order.offset,
                price=trade_price,
                volume=order.volume,
                time=datetime.strftime("%H:%M:%S"),
                gateway_name=self.gateway_name,
                datetime = datetime,
                trading_day = self.trading_day
            )
            trade.sub_strategy_id = order.sub_strategy_id

            self.trades[trade.vt_tradeid] = trade

            # Update spread order.
            spread_order.vt_orderids.append(order.vt_orderid)
            spread_order.status = Status.ALLTRADED

            self.active_spread_orders.pop(spread_order.spread_orderid)

            # Push update to strategy.
            self.strategy.on_order(order)

            self.strategy.pos_dict[spread_order.sub_strategy_id][spread_order.vt_symbol2] += pos_change
            self.strategy.on_trade(trade)
            self.strategy.on_spread_order(spread_order)

    def set_start_date(self, start_date: date):
        self.trading_start_date = int(start_date.strftime("%Y%m%d"))
        self.bar_callback = self.strategy.on_bar

    def load_bar(
        self, start_date: date, interval: Interval, callback: Callable
    ):
        """"""
        if self.tick_callback is not None:
            raise RuntimeError("this function cannot called more than once")

        self.trading_start_date = int(start_date.strftime("%Y%m%d"))
        self.bar_callback = callback

    def load_tick(self, start_date: date, callback: Callable):
        """"""
        if self.bar_callback is not None:
            raise RuntimeError("this function cannot called more than once")

        self.trading_start_date = int(start_date.strftime("%Y%m%d"))
        self.tick_callback = callback

    def send_order(
        self,
        strategy: CtaTemplate,
        direction: Direction,
        offset: Offset,
        price: float,
        volume: float,
        stop: bool,
        lock: bool,
        vt_symbol: str = None,
        sub_strategy_id = None
    ):
        """"""
        price = round_to(price, self.pricetick)
        if stop:
            vt_orderid = self.send_stop_order(direction, offset, price, volume, vt_symbol, sub_strategy_id)
        else:
            vt_orderid = self.send_limit_order(direction, offset, price, volume, vt_symbol, sub_strategy_id)
        return [vt_orderid]

    def send_spread_order(
        self,
        strategy: CtaTemplate,
        direction: Direction,
        offset: Offset,
        price: float,
        volume: float,
        lock: bool,
        vt_symbol1: str,
        vt_symbol2: str,
        sub_strategy_id = None
    ):
        """"""
        self.spread_order_count += 1

        spread_order = SpreadOrder(
            vt_symbol1=vt_symbol1,
            vt_symbol2=vt_symbol2,
            direction=direction,
            offset=offset,
            price=price,
            volume=volume,
            lock=lock,
            spread_orderid=f"{SPREADORDER_PREFIX}.{self.spread_order_count}",
            strategy_name=self.strategy.strategy_name,
        )

        spread_order.sub_strategy_id = sub_strategy_id

        self.active_spread_orders[spread_order.spread_orderid] = spread_order
        self.spread_orders[spread_order.spread_orderid] = spread_order

        return spread_order.spread_orderid

    def send_stop_order(
        self, 
        direction: Direction, 
        offset: Offset, 
        price: float, 
        volume: float,
        vt_symbol: str = None,
        sub_strategy_id = None
    ):
        """"""
        self.stop_order_count += 1

        if vt_symbol is None:
            order_symbol = self.vt_symbols[0]
        else:
            order_symbol = vt_symbol

        stop_order = StopOrder(
            vt_symbol=order_symbol,
            direction=direction,
            offset=offset,
            price=price,
            volume=volume,
            stop_orderid=f"{STOPORDER_PREFIX}.{self.stop_order_count}",
            strategy_name=self.strategy.strategy_name,
        )
        stop_order.sub_strategy_id = sub_strategy_id

        self.active_stop_orders[stop_order.stop_orderid] = stop_order
        self.stop_orders[stop_order.stop_orderid] = stop_order

        return stop_order.stop_orderid

    def send_limit_order(
        self, 
        direction: Direction,
        offset: Offset,
        price: float, 
        volume: float,
        vt_symbol: str = None,
        sub_strategy_id = None
    ):
        """"""
        self.limit_order_count += 1

        if vt_symbol is None:
            order_symbol = self.symbols[0]
            order_exchange = self.exchanges[0]
        else:
            order_symbol, exchange_str = vt_symbol.split(".")
            order_exchange = Exchange(exchange_str)
        
        order = OrderData(
            symbol=order_symbol,
            exchange=order_exchange,
            orderid=str(self.limit_order_count),
            direction=direction,
            offset=offset,
            price=price,
            volume=volume,
            status=Status.SUBMITTING,
            gateway_name=self.gateway_name,
        )
        order.datetime = datetime.fromtimestamp(self.timestamp)
        order.trading_day = self.trading_day
        order.sub_strategy_id = sub_strategy_id

        self.active_limit_orders[order.vt_orderid] = order
        self.limit_orders[order.vt_orderid] = order

        return order.vt_orderid

    def cancel_order(self, strategy: CtaTemplate, vt_orderid: str):
        """
        Cancel order by vt_orderid.
        """
        if vt_orderid.startswith(STOPORDER_PREFIX):
            self.cancel_stop_order(strategy, vt_orderid)
        elif vt_orderid.startswith(SPREADORDER_PREFIX):
            self.cancel_spread_order(strategy, vt_orderid)
        else:
            self.cancel_limit_order(strategy, vt_orderid)

    def cancel_stop_order(self, strategy: CtaTemplate, vt_orderid: str):
        """"""
        if vt_orderid not in self.active_stop_orders:
            return
        stop_order = self.active_stop_orders.pop(vt_orderid)

        stop_order.status = StopOrderStatus.CANCELLED
        self.strategy.on_stop_order(stop_order)

    def cancel_limit_order(self, strategy: CtaTemplate, vt_orderid: str):
        """"""
        if vt_orderid not in self.active_limit_orders:
            return
        order = self.active_limit_orders.pop(vt_orderid)

        order.status = Status.CANCELLED
        self.strategy.on_order(order)

    def cancel_spread_order(self, strategy: CtaTemplate, vt_orderid: str):
        """"""
        if vt_orderid not in self.active_spread_orders:
            return
        order = self.active_spread_orders.pop(vt_orderid)

        order.status = Status.CANCELLED
        self.strategy.on_spread_order(order)

    def cancel_all(self, strategy: CtaTemplate, sub_id = None):
        """
        Cancel all orders, both limit and stop.
        """
        vt_orderids = list(self.active_limit_orders.keys())
        for vt_orderid in vt_orderids:
            if self.active_limit_orders[vt_orderid].sub_strategy_id == sub_id:
                self.cancel_limit_order(strategy, vt_orderid)

        stop_orderids = list(self.active_stop_orders.keys())
        for vt_orderid in stop_orderids:
            if self.active_stop_orders[vt_orderid].sub_strategy_id == sub_id:
                self.cancel_stop_order(strategy, vt_orderid)

        spread_orderids = list(self.active_spread_orders.keys())
        for vt_orderid in spread_orderids:
            if self.active_spread_orders[vt_orderid].sub_strategy_id == sub_id:
                self.cancel_spread_order(strategy, vt_orderid)

    def write_log(self, msg: str, strategy: CtaTemplate = None):
        """
        Write log message.
        """
        if self.timestamp is None:
            msg = f"{msg}"
        else:
            msg = f"{datetime.fromtimestamp(self.timestamp)}\t{msg}"
        self.logs.append(msg)
    
    def send_email(self, msg: str, strategy: CtaTemplate = None):
        """
        Send email to default receiver.
        """
        pass
    
    def sync_strategy_data(self, strategy: CtaTemplate):
        """
        Sync strategy data into json file.
        """
        pass

    def get_engine_type(self):
        """
        Return engine type.
        """
        return self.engine_type

    def put_strategy_event(self, strategy: CtaTemplate):
        """
        Put an event to update strategy status.
        """
        pass

    @staticmethod
    def output(msg):
        """
        Output message of backtesting engine.
        """
        print(f"{datetime.now()}\t{msg}")

    def get_all_trades(self):
        """
        Return all trade data of current backtesting result.
        """
        return list(self.trades.values())

    def get_all_orders(self):
        """
        Return all limit order data of current backtesting result.
        """
        return list(self.limit_orders.values())

    def get_all_daily_results(self):
        """
        Return all daily result data.
        """
        return list(self.daily_results.values())


class DailyResult:
    """"""

    def __init__(self, date: date):
        """"""
        self.date = date
        self.close_price_dict: Dict(str, float) = dict()
        self.pre_close_dict: Dict(str, float) = dict()

        self.trades = []
        self.trade_count = 0

        self.start_pos_dict = dict()
        self.end_pos_dict = dict()

        self.turnover = 0
        self.commission = 0
        self.slippage = 0

        self.trading_pnl = 0
        self.holding_pnl = 0
        self.total_pnl = 0
        self.net_pnl = 0

    @staticmethod
    def merge_dict(dict1, dict2):
        ''' Merge dictionaries and keep values of common keys in list'''
        dict3 = {**dict1, **dict2}
        for key, value in dict3.items():
            if key in dict1 and key in dict2:
                    dict3[key] = value + dict1[key]
        
        return dict3

    def combine_result(self, other):
        if self.date != other.date:
            raise RuntimeError("can only combine result in same day")
        self.close_price_dict.update(other.close_price_dict)
        self.pre_close_dict.update(other.pre_close_dict)
        self.trades.extend(other.trades)
        self.trades.sort(key=lambda x: x.datetime)
        self.trade_count += other.trade_count
        self.start_pos_dict = self.merge_dict(self.start_pos_dict, other.start_pos_dict)
        self.end_pos_dict = self.merge_dict(self.end_pos_dict, other.end_pos_dict)
        self.turnover += other.turnover
        self.commission += other.commission
        self.slippage += other.slippage
        self.trading_pnl += other.trading_pnl
        self.holding_pnl += other.holding_pnl
        self.total_pnl += other.holding_pnl
        self.net_pnl += other.net_pnl 

    def add_trade(self, trade: TradeData):
        """"""
        self.trades.append(trade)

    def calculate_pnl(
        self,
        vt_symbol: str,
        pre_close: float,
        start_pos: float,
        size: int,
        rate: CommissionRate,
        slippage: float,
    ):
        """"""
        self.pre_close_dict[vt_symbol] = pre_close

        # Holding pnl is the pnl from holding position at day start
        self.start_pos_dict[vt_symbol] = start_pos
        self.end_pos_dict[vt_symbol] = start_pos
        self.holding_pnl += self.start_pos_dict[vt_symbol] * \
            (self.close_price_dict[vt_symbol] - self.pre_close_dict[vt_symbol]) * size

        # Trading pnl is the pnl from new trade during the day
        self.trade_count = len(self.trades)

        long_today_pos = 0
        short_today_pos = 0
        close_today_volume = 0

        for trade in self.trades:
            if trade.vt_symbol != vt_symbol:
                continue
            if (trade.direction == Direction.LONG and trade.offset == Offset.OPEN) or (trade.direction == Direction.SHORT and trade.offset != Offset.OPEN):
                pos_change = trade.volume
            else:
                pos_change = -trade.volume

            if (trade.offset == Offset.OPEN):
                if trade.direction == Direction.LONG:
                    long_today_pos += trade.volume
                else:
                    short_today_pos += trade.volume
            else:
                if trade.direction == Direction.LONG:
                    close_today_volume = min(long_today_pos, trade.volume)
                    long_today_pos = max(0, long_today_pos - trade.volume)
                else:
                    close_today_volume = min(short_today_pos, trade.volume)
                    short_today_pos = max(0, short_today_pos - trade.volume)

            turnover = trade.price * trade.volume * size

            self.trading_pnl += pos_change * \
                (self.close_price_dict[vt_symbol] - trade.price) * size
            self.end_pos_dict[vt_symbol] += pos_change
            self.turnover += turnover
            if trade.offset == Offset.OPEN:
                self.commission += turnover * rate.open_commission_rate + trade.volume * rate.open_commission_amount
            else:
                close_today_turnover = close_today_volume * trade.price * size
                self.commission += (turnover - close_today_turnover) * rate.close_commission_rate + (trade.volume - close_today_volume) * rate.close_commission_amount + \
                    close_today_turnover * rate.close_today_commission_rate + close_today_volume * rate.close_today_commission_amount
            self.slippage += trade.volume * size * slippage

        # Net pnl takes account of commission and slippage cost
        self.total_pnl = self.trading_pnl + self.holding_pnl
        self.net_pnl = self.total_pnl - self.commission - self.slippage

g_history_data = None

def init_process(history_data):
    global g_history_data

    g_history_data = history_data

def optimize(
    target_name: str,
    strategy_class: CtaTemplate,
    setting: dict,
    vt_symbols: List[str],
    interval: Interval,
    start: datetime,
    rate: float,
    slippage: float,
    size: float,
    pricetick: float,
    capital: int,
    end: datetime,
    mode: BacktestingMode,
    start_date: date = None,
    history_bar_data = None,
    iterate_mode: IterateMode = IterateMode.NONE,
    parallel: int = 1,
    active_arb_mode : bool = False
):
    """
    Function for running in multiprocessing.pool
    """
    from strategies.combine_strategy import CombineStrategy

    engine = BacktestingEngine()
    
    engine.set_parameters(
        vt_symbols=vt_symbols,
        interval=interval,
        start=start,
        rate=rate,
        slippage=slippage,
        size=size,
        pricetick=pricetick,
        capital=capital,
        end=end,
        mode=mode,
        iterate_mode=iterate_mode,
        active_arb_mode=active_arb_mode
    )

    if parallel > 1:
        engine.add_strategy(CombineStrategy, {})
        for i in range(parallel):
            engine.strategy.add_strategy(strategy_class, strategy_class.__name__, engine.vt_symbols, setting)
    else:
        engine.add_strategy(strategy_class, setting)
    if start_date:
        engine.set_start_date(start_date)
    if g_history_data is not None:
        engine.output("从共享内存中读取历史数据")
        engine.history_data = g_history_data
        engine.history_bar_data = history_bar_data
        engine.output("从共享内存中读取历史数据完毕")
    else:
        engine.load_data()
    engine.run_backtesting()
    engine.calculate_result()
    engine.strategy.on_stop()
    statistics = engine.calculate_statistics(output=False)

    target_value = statistics[target_name]
    return (str(setting), target_value, statistics, engine.daily_results, engine.trades, setting, engine.daily_df)


@lru_cache(maxsize=1000000)
def _ga_optimize(parameter_values: tuple):
    """"""
    setting = dict(parameter_values)

    result = optimize(
        ga_target_name,
        ga_strategy_class,
        setting,
        ga_vt_symbols,
        ga_interval,
        ga_start,
        ga_rate,
        ga_slippage,
        ga_size,
        ga_pricetick,
        ga_capital,
        ga_end,
        ga_mode
    )
    return (result[1],)


def ga_optimize(parameter_values: list):
    """"""
    return _ga_optimize(tuple(parameter_values))


# @lru_cache(maxsize=999)
def load_bar_data(
    symbol: str,
    exchange: Exchange,
    interval: Interval,
    start: datetime,
    end: datetime
):
    """"""
    return database_manager.load_bar_data(
        symbol, exchange, interval, start, end
    )


# @lru_cache(maxsize=999)
def load_tick_data(
    symbol: str,
    exchange: Exchange,
    start: datetime,
    end: datetime
):
    """"""
    return database_manager.load_tick_data(
        symbol, exchange, start, end
    )

def load_tick_data_with_millsec_adjust(
    symbol: str,
    exchange: Exchange,
    start: datetime,
    end: datetime
):
    ticks = load_tick_data(symbol, exchange, start, end)
    if exchange == Exchange.CZCE and len(ticks) > 0:
        prev_sec = 0.
        same_sec_num = 0
        for tick in ticks:
            mill, sec = math.modf(tick.data.tick.timestamp)
            if prev_sec > 0.0 and math.isclose(prev_sec, sec, rel_tol=0, abs_tol=1e-6) and math.isclose(mill, 0):
                same_sec_num += 1
                tick.data.tick.timestamp = sec + same_sec_num * 0.1
            elif not math.isclose(sec, prev_sec, rel_tol=0, abs_tol=1e-6):
                same_sec_num = 0
            prev_sec = sec
    return ticks



def load_engine_data_in_process(q, start, end, symbols, exchanges, mode, interval, active_arb_mode=False):
    history_bar_data, history_data = load_engine_data(start, end, symbols, exchanges, mode, interval, active_arb_mode)
    mmap_file = q[0]

    fd = os.open(mmap_file, os.O_RDWR)
    trans_map = mmap.mmap(fd, 0)
    trans_map.resize(ctypes.sizeof(CTickBarData) * len(history_data))

    pointer = ctypes.pointer(CTickBarData.from_buffer(trans_map))
    for index, data in enumerate(history_data):
        pointer[index] = data
    
    q.append(history_bar_data)
    q.append(len(history_data))

    del pointer
    trans_map.close()
    del trans_map
    os.close(fd)


def next_month_start(start: datetime, end: datetime):
    next_month = date(start.year, start.month, 1) + timedelta(days=40)
    next_month = datetime.combine(date(next_month.year, next_month.month, 1), start.time())
    return min(next_month, end)

def convert_to_arbBar_data(symbols, exchanges, interval, tick_data, bar_data, active_order_mode):
    trading_days = []
    minutes_dict = {}
    arbBar_data = []
    for bar in bar_data[symbols[0]]:
        trading_day = bar.trading_day
        if trading_day not in trading_days:
            trading_days.append(trading_day)
            minutes_dict[trading_day] = []
        minutes_dict[trading_day].append(bar.timestamp)

    consecutive_ticks = 2
    if exchanges[0] == Exchange.SHFE or active_order_mode:
        consecutive_ticks = 4
    index1 = 0
    index2 = 0
    tick1 = tick_data[symbols[0]]
    tick2 = tick_data[symbols[1]]
    tick1_len = len(tick1) - 1
    tick2_len = len(tick2) - 1
    for trading_day in trading_days:
        while True:
            if index1 < tick1_len and tick1[index1].trading_day < trading_day:
                index1 += 1
                continue
            if index2 < tick2_len and tick2[index2].trading_day < trading_day:
                index2 += 1
                continue
            break
        ask_list = []
        bid_list = []
        mid_list = []
        
        last_equal_timestamp = 0
        last_equal_times = 0
        while (index1 < tick1_len and tick1[index1].trading_day == trading_day) or (index2 < tick2_len and tick2[index2].trading_day == trading_day):
            if tick1[index1].ask_volume_1 > 0 and tick2[index2].bid_volume_1 > 0 and tick1[index1].trading_day == trading_day and tick2[index2].trading_day == trading_day:
                ask_list.append((max(tick1[index1].timestamp, tick2[index2].timestamp), tick1[index1].ask_price_1 - tick2[index2].bid_price_1))
            if tick1[index1].bid_volume_1 > 0 and tick2[index2].ask_volume_1 > 0 and tick1[index1].trading_day == trading_day and tick2[index2].trading_day == trading_day:
                bid_list.append((max(tick1[index1].timestamp, tick2[index2].timestamp),tick1[index1].bid_price_1 - tick2[index2].ask_price_1))
            if tick1[index1].ask_volume_1 > 0 and tick1[index1].bid_volume_1 > 0 and tick2[index2].ask_volume_1 > 0 and  tick2[index2].bid_volume_1 > 0 and tick1[index1].trading_day == trading_day and tick2[index2].trading_day == trading_day:
                mid_list.append((max(tick1[index1].timestamp, tick2[index2].timestamp), ((tick1[index1].bid_price_1 + tick1[index1].ask_price_1) - (tick2[index2].bid_price_1 + tick2[index2].ask_price_1))/2))

            if index2 == tick2_len:
                index1 += 1
            elif index1 == tick1_len:
                index2 += 1
            elif tick1[index1 + 1].timestamp < tick2[index2+1].timestamp:
                index1 += 1
            elif tick1[index1+1].timestamp > tick2[index2+1].timestamp:
                index2 += 1
            elif tick1[index1+1].timestamp == tick2[index2+1].timestamp and tick1[index1+1].timestamp == last_equal_timestamp:
                if last_equal_times % 2  == 0:
                    index1 += 1
                else:
                    index2 += 1
                last_equal_times += 1
            elif tick1[index1+1].timestamp == tick2[index2+1].timestamp and tick1[index1+1].timestamp != last_equal_timestamp:
                last_equal_timestamp = tick1[index1+1].timestamp
                last_equal_times = 1
                index1 += 1

        ask_list2 = []
        bid_list2 = []
        mid_list2 = mid_list
        for i in range(len(ask_list) - consecutive_ticks + 1):
            ask_list2.append((ask_list[i][0], max(ask_list[i:i+consecutive_ticks], key=lambda x: x[1])[1]))
        for i in range(len(bid_list) - consecutive_ticks + 1):
            bid_list2.append((bid_list[i][0], min(bid_list[i:i+consecutive_ticks], key=lambda x: x[1])[1]))

        trading_day_arb_data = list()
        start_timestamp = minutes_dict[trading_day][0] - int(bar_data[symbols[0]][0].interval.decode()[:-1]) * 60 + 2
        for bar in minutes_dict[trading_day]:
            arbBar = CArbBarData()
            arbBar.symbol1 = symbols[0].encode('utf-8')
            arbBar.symbol2 = symbols[1].encode('utf-8')
            arbBar.exchange1 = exchanges[0].value.encode('utf-8')
            arbBar.exchange2 = exchanges[1].value.encode('utf-8')
            arbBar.trading_day = trading_day
            arbBar.interval = interval.value.encode('utf-8')
            arbBar.bid_ticks = arbBar.ask_ticks = arbBar.mid_ticks = 0
            arbBar.bid_open_price = arbBar.bid_high_price = arbBar.bid_low_price = arbBar.bid_close_price = 0
            arbBar.ask_open_price = arbBar.ask_high_price = arbBar.ask_low_price = arbBar.ask_close_price = 0
            arbBar.mid_open_price = arbBar.mid_high_price = arbBar.mid_low_price = arbBar.mid_close_price = 0
            arbBar.timestamp = bar
            arbBar.vt_symbol1 = (symbols[0] + "." + exchanges[0].value).encode('utf-8')
            arbBar.vt_symbol2 = (symbols[1] + "." + exchanges[1].value).encode('utf-8')
            arbBar.gateway_name = bar_data[symbols[0]][0].gateway_name
            trading_day_arb_data.append(arbBar)

        arb_index = 0
        for ask_elem in ask_list2:
            if ask_elem[0] < start_timestamp:
                continue
            while arb_index < len(trading_day_arb_data) and trading_day_arb_data[arb_index].timestamp <= ask_elem[0]:
                arb_index += 1
            if arb_index >= len(trading_day_arb_data):
                break
            arbBar = trading_day_arb_data[arb_index]
            if arbBar.ask_ticks == 0:
                arbBar.ask_open_price = arbBar.ask_high_price = arbBar.ask_low_price = arbBar.ask_close_price = ask_elem[1]
            else:
                arbBar.ask_high_price = max(arbBar.ask_high_price, ask_elem[1])
                arbBar.ask_low_price = min(arbBar.ask_low_price, ask_elem[1])
                arbBar.ask_close_price = ask_elem[1]
            arbBar.ask_ticks += 1
        
        arb_index = 0
        for bid_elem in bid_list2:
            if bid_elem[0] < start_timestamp:
                continue
            while arb_index < len(trading_day_arb_data) and trading_day_arb_data[arb_index].timestamp <= bid_elem[0]:
                arb_index += 1
            if arb_index >= len(trading_day_arb_data):
                break
            arbBar = trading_day_arb_data[arb_index]
            if arbBar.bid_ticks == 0:
                arbBar.bid_open_price = arbBar.bid_high_price = arbBar.bid_low_price = arbBar.bid_close_price = bid_elem[1]
            else:
                arbBar.bid_high_price = max(arbBar.bid_high_price, bid_elem[1])
                arbBar.bid_low_price = min(arbBar.bid_low_price, bid_elem[1])
                arbBar.bid_close_price = bid_elem[1]
            arbBar.bid_ticks += 1

        arb_index = 0
        for mid_elem in mid_list2:
            if mid_elem[0] < start_timestamp:
                continue
            while arb_index < len(trading_day_arb_data) and trading_day_arb_data[arb_index].timestamp <= mid_elem[0]:
                arb_index += 1
            if arb_index >= len(trading_day_arb_data):
                break
            arbBar = trading_day_arb_data[arb_index]
            if arbBar.mid_ticks == 0:
                arbBar.mid_open_price = arbBar.mid_high_price = arbBar.mid_low_price = arbBar.mid_close_price = mid_elem[1]
            else:
                arbBar.mid_high_price = max(arbBar.mid_high_price, mid_elem[1])
                arbBar.mid_low_price = min(arbBar.mid_low_price, mid_elem[1])
                arbBar.mid_close_price = mid_elem[1]
            arbBar.mid_ticks += 1
        
        for arb_elem in trading_day_arb_data:
            data = CTickBarData()
            data.type = b"arbBar"
            data.data.arbBar = arb_elem
            arbBar_data.append(data)

    return arbBar_data

# @lru_cache(maxsize=999)
def load_engine_data(start, end, symbols, exchanges, mode, interval, active_order_mode=False):
    # Load 30 days of data each time and allow for progress update
    # progress_delta = timedelta(days=30)
    total_delta = end - start

    start1 = start
    # end1 = start + progress_delta
    end1 = next_month_start(start1, end)
    progress = 0

    history_data = []
    history_bar_data = dict()
    history_tick_data = dict()

    while start1 < end:
        end1 = min(end1, end)  # Make sure end time stays within set range
        
        for index, symbol in enumerate(symbols):
            exchange = exchanges[index]
            if mode == BacktestingMode.BAR:
                data = load_bar_data(
                    symbol,
                    exchange,
                    interval,
                    start1,
                    end1
                )
                history_data.extend(data)
                if symbol not in history_bar_data:
                    history_bar_data[symbol] = list()
                history_bar_data[symbol].extend([x.data.bar for x in data])
                history_bar_data[symbol].sort(key=lambda x: x.timestamp)
                del data
            elif mode == BacktestingMode.TICK:
                data = load_tick_data_with_millsec_adjust(
                    symbol,
                    exchange,
                    start1,
                    end1
                )
                history_data.extend(data)
                del data
            elif mode == BacktestingMode.MIXED:
                data = load_bar_data(
                    symbol,
                    exchange,
                    interval,
                    start1,
                    end1
                )
                history_data.extend(data)
                if symbol not in history_bar_data:
                    history_bar_data[symbol] = list()
                history_bar_data[symbol].extend([x.data.bar for x in data])
                history_bar_data[symbol].sort(key=lambda x: x.timestamp)
                del data
                data = load_tick_data_with_millsec_adjust(
                    symbol,
                    exchange,
                    start1,
                    end1
                )
                history_data.extend(data)
                del data
            else:
                bar_data = load_bar_data(
                    symbol,
                    exchange,
                    interval,
                    start1,
                    end1
                )
                history_data.extend(bar_data)
                if symbol not in history_bar_data:
                    history_bar_data[symbol] = list()
                history_bar_data[symbol].extend([x.data.bar for x in bar_data])
                history_bar_data[symbol].sort(key=lambda x: x.timestamp)
                data = load_tick_data_with_millsec_adjust(
                    symbol,
                    exchange,
                    start1,
                    end1
                )
                if symbol not in history_tick_data:
                    history_tick_data[symbol] = list()
                history_tick_data[symbol].extend([x.data.tick for x in data])
                # history_data.extend(data)
                del data
                del bar_data
        
        # progress += progress_delta / total_delta
        progress += (end1 - start1) / total_delta

        progress = min(progress, 1)
        progress_bar = "#" * int(progress * 10)
        BacktestingEngine.output(f"加载进度：{progress_bar} [{progress:.0%}]")
        
        start1 = end1
        # end1 += progress_delta
        end1 = next_month_start(start1, end)

    if mode == BacktestingMode.ArbMIXED:
        history_data.extend(convert_to_arbBar_data(symbols, exchanges, interval, history_tick_data, history_bar_data, active_order_mode))

    del history_tick_data

    history_data.sort(key=lambda x: x.data.tick.timestamp if x.type == b"tick" else (x.data.bar.timestamp if x.type == b'bar' else x.data.arbBar.timestamp + 1))
    return history_bar_data, history_data


# GA related global value
ga_end = None
ga_mode = None
ga_target_name = None
ga_strategy_class = None
ga_setting = None
ga_vt_symbols = None
ga_interval = None
ga_start = None
ga_rate = None
ga_slippage = None
ga_size = None
ga_pricetick = None
ga_capital = None
